package com.zf.nospring.kafka.simple;

import com.zf.nospring.kafka.utils.KafkaUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.junit.jupiter.api.Test;

import java.time.Duration;
import java.util.ArrayList;

/**
 * 使用KafkaConsumer消费Topic中的消息
 */
public class ConsumerTest {

    @Test
    public void testConsumer() {

        // 创建消费者对象
        KafkaConsumer<String, String> kafkaConsumer = KafkaUtils.getDefaultConsumer();
        // 注册要消费的主题（可以消费多个主题）
        ArrayList<String> topics = new ArrayList<>();
        topics.add("test");
        kafkaConsumer.subscribe(topics);
        try {
            while (true) {
                // 获取消息，超时时间设置为1s, 1s后如果没有获取到消息则返回空集
                ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
                // 打印消费到的数据
                for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                    String topic = consumerRecord.topic();
                    String msg = consumerRecord.value();
                    long offset = consumerRecord.offset();
                    int partition = consumerRecord.partition();
                    System.out.printf("consumerRecord: %s\ntopic: %s, msg: %s, offset: %d, partition: %d\n", consumerRecord, topic, msg, offset, partition);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            kafkaConsumer.close();
        }

    }
}
